package com.shujia.stream

import com.shujia.util.FlinkTableEnv
import org.apache.flink.table.api.TableEnvironment

object CardAvgSpeedAndFlow {
  def main(args: Array[String]): Unit = {

    val table: TableEnvironment = FlinkTableEnv.getTableEnv(5000)



    /**
      *
      * 创建source表
      *
      */

    table.executeSql(
      """
        |CREATE TABLE cars_kafka (
        |  `car` STRING,
        |  `city_code` STRING,
        |  `county_code` STRING,
        |  `card` BIGINT,
        |  `camera_id` STRING,
        |  `orientation` STRING,
        |  `road_id` BIGINT,
        |  `time` BIGINT,
        |  `speed`  DOUBLE,
        |  `ts` AS TO_TIMESTAMP(FROM_UNIXTIME(`time`,'yyyy-MM-dd HH:mm:ss')), -- 事件事件时间字段
        |  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
        |) WITH (
        |  'connector' = 'kafka',
        |  'topic' = 'cars',
        |  'properties.bootstrap.servers' = 'master:9092',
        |  'properties.group.id' = 'CardAvgSpeedAndFlow001',
        |  'scan.startup.mode' = 'earliest-offset',
        |  'format' = 'json',
        |  'json.ignore-parse-errors' = 'true'
        |)
        |
      """.stripMargin)

    table.executeSql(
      """
        |CREATE TABLE card_avg_speed_and_flow (
        |  card BIGINT,
        |  window_start TIMESTAMP(3),
        |  window_end TIMESTAMP(3),
        |  avg_speed DOUBLE,
        |  flow BIGINT,
        |  PRIMARY KEY (card,window_start,window_end) NOT ENFORCED
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/car',
        |   'table-name' = 'card_avg_speed_and_flow',
        |   'username' = 'root',
        |   'password' = '123456'
        |)
        |
      """.stripMargin)


    /**
      * 统计卡口的平均车速和车流量，每隔1分钟统计一次，统计最近5分钟的车辆
      *
      */
    table.executeSql(
      """
        |insert into card_avg_speed_and_flow
        |select card,window_start, window_end,avg(speed) as avg_speed,count(1) flow from
        |TABLE(
        |    HOP(TABLE cars_kafka, DESCRIPTOR(ts), INTERVAL '1' MINUTES, INTERVAL '5' MINUTES)
        |    )
        |group by card,window_start,window_end
        |
      """.stripMargin)


  }

}
